package com.bd08.flink.demo.zhang.table


import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.Expressions.{$, row}
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.api.{DataTypes, EnvironmentSettings, TableEnvironment}

import scala.tools.nsc.doc.base.comment.Row

object TableDemo extends App {
  //创建Table执行上下文
  val benv = EnvironmentSettings.inStreamingMode()
  val tenv = TableEnvironment.create(benv)
  //创建流到table执行上下文
  val env = StreamExecutionEnvironment.getExecutionEnvironment
  val stenv = StreamTableEnvironment.create(env)

  //获取数据
  val t1=tenv.fromValues(DataTypes.ROW(
    DataTypes.FIELD("name",DataTypes.STRING().notNull()),
    DataTypes.FIELD("age",DataTypes.STRING().notNull()),
    DataTypes.FIELD("birth",DataTypes.STRING().notNull())
  ),row("Jason","9","2010-1-1"),row("Tom","9","2010-1-1"))
//  val users= tenv.fromDataSet(ds)
  t1.printSchema()
  t1.select($("*")).execute().print()
  tenv.createTemporaryView("person",t1)
  tenv.executeSql("select * from person").print()
//  users.first(10).print()
//  users.select("_1,_2").toDataSet[Row].print()


}
